package com.sunline.poc.oozie.service.impl;

import com.sunline.poc.oozie.ApplicationProperties;
import com.sunline.poc.oozie.client.OozieClient;
import com.sunline.poc.oozie.client.OozieDTO;
import com.sunline.poc.oozie.config.ApplicationException;
import com.sunline.poc.oozie.config.LoginUserContext;
import com.sunline.poc.oozie.dao.OozieDao;
import com.sunline.poc.oozie.service.OozieService;
import com.sunline.poc.oozie.vo.UserInfoVO;
import com.sunline.poc.oozie.vo.OozieJobConfigVO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

@Service
public class OozieServiceImpl implements OozieService {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private OozieClient oozieClient;

    @Autowired
    private OozieDao oozieDao;

    @Autowired
    private ApplicationProperties applicationProperties;

    private Map<String,OozieDTO> jobInfos = new HashMap<>();


    @Override
    public OozieDTO start(String jobName) throws ApplicationException {
        Properties properties = getStartProperties(jobName);
        String jobId =  oozieClient.start(properties);
        OozieDTO dto = new OozieDTO();
        dto.setJobName(jobName);
        dto.setJobId(jobId);
        dto.setJobStatus(oozieClient.status(jobId));
        dto.setOwner(LoginUserContext.getLoginUser().getUserName());
        jobInfos.put(jobId,dto);
        return dto;
    }

    @Override
    public boolean kill(String jobId) {
        OozieDTO dto = jobInfos.get(jobId);
        String loginUserName  = LoginUserContext.getLoginUser().getUserName();
        if (dto == null){
            throw new RuntimeException("job not found.");
        }

        if (!loginUserName.equalsIgnoreCase(dto.getOwner())){
            throw new RuntimeException("you have no right to do it. this job's owner is " + dto.getOwner());
        }

        return oozieClient.kill(jobId);
    }

    @Override
    public OozieDTO restart(String jobId,String jobName) {
        throw new RuntimeException("没有实现");
    }

    @Override
    public OozieDTO getJobInfos(String jobName) {
        return null;
    }

    @Override
    public String getJobStatus(String jobId) {
        OozieDTO dto = jobInfos.get(jobId);
        String loginUserName  = LoginUserContext.getLoginUser().getUserName();
        if (dto == null){
            throw new RuntimeException("job not exists/run");
        }
        if (!loginUserName.equalsIgnoreCase(dto.getOwner())){
            throw new RuntimeException("you have no right to do it. this job's owner is " + dto.getOwner());
        }
        return oozieClient.status(jobId);
    }

    @Override
    public OozieDTO rerun(String jobName, String etlDate) throws ApplicationException {
        OozieJobConfigVO jobConfigVO = oozieDao.findOozieJobConfig(jobName);
        UserInfoVO loginuser =  LoginUserContext.getLoginUser();

        if (null == jobConfigVO){
            throw new RuntimeException("job not define");
        }

        Properties properties = getStartProperties(jobName);
        properties.setProperty("clientName",loginuser.getUserName());
        properties.setProperty("dtTag",etlDate);
        String jobId =  oozieClient.start(properties);
        OozieDTO dto = new OozieDTO();
        dto.setJobName(jobName);
        dto.setJobId(jobId);
        dto.setJobStatus(oozieClient.status(jobId));
        dto.setOwner(LoginUserContext.getLoginUser().getUserName());
        jobInfos.put(jobId,dto);
        return dto;

    }


    private  void setProperty(Properties properties,String key,String value){
        if (!StringUtils.isEmpty(value)){
            properties.setProperty(key,value);
        }
    }

    private Properties getStartProperties(String jobName){
        OozieJobConfigVO jobConfigVO = oozieDao.findOozieJobConfig(jobName);
        UserInfoVO loginuser =  LoginUserContext.getLoginUser();

        if (null == jobConfigVO){
            throw new RuntimeException("job not define");
        }
        Properties properties = new Properties();

        setProperty(properties,"nameNode", applicationProperties.getHdfsNameNodeURI());
        setProperty(properties,"jobTracker", applicationProperties.getJobTracker());


        Map args =  jobConfigVO.getProperties();
        if (args != null){
            Iterator<Map.Entry<String, String>> entries = args.entrySet().iterator();
            while (entries.hasNext()) {
                Map.Entry<String, String> entry = entries.next();
                properties.put(entry.getKey(),entry.getValue());
            }
        }
        return  properties;
    }
}
